/*
 *  Copyright (c) 1998, Regents of the University of California
 *  All rights reserved.
 *  Redistribution and use in source and binary forms, with or without
 *  modification, are permitted provided that the following conditions are met:
 *
 *      * Redistributions of source code must retain the above copyright
 *        notice, this list of conditions and the following disclaimer.
 *      * Redistributions in binary form must reproduce the above copyright
 *        notice, this list of conditions and the following disclaimer in the
 *        documentation and/or other materials provided with the distribution.
 *      * Neither the name of the University of California, Berkeley nor the
 *        names of its contributors may be used to endorse or promote products
 *        derived from this software without specific prior written permission.
 *
 *  THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND ANY
 *  EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 *  WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 *  DISCLAIMED. IN NO EVENT SHALL THE REGENTS AND CONTRIBUTORS BE LIABLE FOR ANY
 *  DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 *  (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 *  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 *  ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 *  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 *  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
package org.demis.elf.dbreader;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.StringTokenizer;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.velocity.Template;
import org.apache.velocity.app.VelocityEngine;
import org.apache.velocity.exception.ParseErrorException;
import org.apache.velocity.exception.ResourceNotFoundException;
import org.demis.elf.Column;
import org.demis.elf.ColumnSortSequence;
import org.demis.elf.DataLine;
import org.demis.elf.ForeignKey;
import org.demis.elf.Index;
import org.demis.elf.PrimaryKey;
import org.demis.elf.Schema;
import org.demis.elf.Table;
import org.demis.elf.configuration.DatabaseConfiguration;
import org.demis.elf.dbreader.generic.GenericSchemaReader;

/**
 * @author <a href="mailto:demis27@demis27.net">Stéphane kermabon</a>
 * @version 1.0
 */
public abstract class AbstractSchemaReader implements SchemaReader {

    private static final Log logger = LogFactory.getLog(AbstractSchemaReader.class);

    @Override
    public Schema read(String url, String login, String password, String shema) {
        DatabaseConfiguration databaseConfiguration = new DatabaseConfiguration();
        databaseConfiguration.setUrl(url);
        databaseConfiguration.setUser(login);
        databaseConfiguration.setPassword(password);
        databaseConfiguration.setSchemaName(shema);

        return read(databaseConfiguration);

    }

    protected Template initVelocity(String templateName) {
        VelocityEngine ve = new VelocityEngine();
        try {
            Properties p = new Properties();
            p.setProperty("file.resource.loader.class", "org.apache.velocity.runtime.resource.loader.ClasspathResourceLoader");
            ve.init(p);
            Template mappingTemplate = ve.getTemplate(templateName);
            return mappingTemplate;
        } catch (ResourceNotFoundException ex) {
            if (logger.isFatalEnabled()) {
                logger.fatal("Error when initialized velocity", ex);
            }
        } catch (ParseErrorException ex) {
            if (logger.isFatalEnabled()) {
                logger.fatal("Error when initialized velocity", ex);
            }
        } catch (Exception ex) {
            if (logger.isFatalEnabled()) {
                logger.fatal("Error when initialized velocity", ex);
            }
        }
        return null;
    }

    protected abstract String getSchemaNameForMetaData(String schemaName);

    protected abstract String getTableNameForMetaData(String tableName);

    // TODO : read Column IS_AUTOINCREMENT

    @Override
    public Schema read(DatabaseConfiguration configuration) {
        if (logger.isInfoEnabled()) {
            logger.info("read schema with " + this.getClass().getName());
            logger.info("schema   : " + configuration.getSchemaName());
            logger.info("user     : " + configuration.getUser());
            logger.info("password : " + configuration.getPassword());
            logger.info("url      : " + configuration.getUrl());
        }

        Schema schema = readSchema(configuration);
        if (schema == null) {
            return null;
        }
        readTables(configuration, schema);
        readColumns(configuration, schema);
        readPrimaryKeys(configuration, schema);
        readForeignKeys(configuration, schema);
        readIndexes(configuration, schema);
        //readColumnsDefaultValue(configuration, schema);
        //readUniqueConstraint(configuration, schema);

        if (configuration.isReadData() && configuration.getFilter() != null) {
            readData(configuration, schema);
        }

        return schema;
    }

    @Override
    public Schema readSchema(DatabaseConfiguration configuration) {
        Schema schema = null;
        ResultSet resultSet = null;
        Connection connection = null;
        try {
            connection = getConnection(configuration);
            if (logger.isDebugEnabled()) {
                logger.debug("reading schemas, try to found : " + configuration.getSchemaName());
            }
            DatabaseMetaData metaData = connection.getMetaData();
            resultSet = metaData.getSchemas();
            while (resultSet.next()) {
                String schemaFound = resultSet.getString("TABLE_SCHEM");
                if (schemaFound.equalsIgnoreCase(configuration.getSchemaName())) {
                    schema = new Schema(configuration.getSchemaName());
                    if (logger.isInfoEnabled()) {
                        logger.info("read schema : schema #" + schema + " found");
                    }
                } else {
                    if (logger.isDebugEnabled()) {
                        logger.debug("read schema : other schema " + schemaFound + " found");
                    }
                }
            }
            if (logger.isDebugEnabled()) {
                logger.debug("finished read schemas");
            }
        } catch (SQLException ex) {
            if (logger.isErrorEnabled()) {
                logger.error("Error when reading schema " + configuration.getSchemaName(), ex);
            }
        }
        finally {
            cleanConnection(connection, null, resultSet);
        }
        return schema;
    }

    @Override
    public void readTables(DatabaseConfiguration configuration, Schema schema) {
        Map<String, Table> tables = new HashMap<String, Table>();
        Connection connection = null;
        ResultSet resultSet = null;
        try {
            connection = getConnection(configuration);
            DatabaseMetaData metaData = connection.getMetaData();
            // read tables
            if (logger.isInfoEnabled()) {
                logger.info("read tables for schema #" + schema);
            }
            resultSet = metaData.getTables(
                    null,
                    getSchemaNameForMetaData(schema.getName()),
                    null,
                    new String[]{"TABLE"});
            while (resultSet.next()) {
                Table table = new Table(resultSet.getString("TABLE_NAME"));
                table.setRemarks(resultSet.getString("REMARKS"));
                tables.put(table.getName(), table);
                if (logger.isInfoEnabled()) {
                    logger.info("read table #" + table);
                }
            }
        } catch (SQLException ex) {
            if (logger.isErrorEnabled()) {
                logger.error("Error when reading schema's table " + configuration.getSchemaName(), ex);
            }
        }
        finally {
            cleanConnection(connection, null, resultSet);
        }
        schema.addTables(new ArrayList<Table>(tables.values()));
    }

    @Override
    public void readColumns(DatabaseConfiguration configuration, Schema schema) {
        Connection connection = null;
        ResultSet resultSet = null;

        try {
            connection = getConnection(configuration);
            DatabaseMetaData metaData = connection.getMetaData();
            for (Table table : schema.getTables()) {
                if (logger.isInfoEnabled()) {
                    logger.info("read columns for table #" + table + " on schema #" + schema);
                }
                // columns
                resultSet = metaData.getColumns(
                        null,
                        getSchemaNameForMetaData(configuration.getSchemaName()),
                        getTableNameForMetaData(table.getName()),
                        null);
                while (resultSet.next()) {
                    Column column = new Column(table, resultSet.getString("COLUMN_NAME"));
                    // type
                    column.setType(resultSet.getInt("DATA_TYPE"));
                    // column size
                    column.setSize(resultSet.getInt("COLUMN_SIZE"));
                    // scale
                    column.setScale(resultSet.getInt("DECIMAL_DIGITS"));

                    // nullable
                    int nullable = resultSet.getInt("NULLABLE");
                    column.setNotNull(nullable != DatabaseMetaData.columnNullable);
                    // remarks
                    column.setRemarks(resultSet.getString("REMARKS"));
                    // ordinal position
                    int position = resultSet.getInt("ORDINAL_POSITION");
                    column.setPosition(position);

                    table.addColumn(column);
                    if (logger.isInfoEnabled()) {
                        logger.info("read column #" + column);
                    }
                }
                resultSet.close();
            }
        } catch (SQLException ex) {
            if (logger.isErrorEnabled()) {
                logger.error("Error when reading schema's table's column", ex);
            }
        } finally {
            cleanConnection(connection, null, resultSet);
        }
    }

    @Override
    public void readPrimaryKeys(DatabaseConfiguration configuration, Schema schema) {
        Connection connection = null;
        ResultSet resultSet = null;

        try {
            connection = getConnection(configuration);
            DatabaseMetaData metaData = connection.getMetaData();
            for (Table table : schema.getTables()) {
                // primary key
                resultSet = metaData.getPrimaryKeys(
                        null,
                        getSchemaNameForMetaData(configuration.getSchemaName()),
                        getTableNameForMetaData(table.getName()));
                PrimaryKey primaryKey = new PrimaryKey();
                while (resultSet.next()) {
                    Column column = table.getColumn(resultSet.getString("COLUMN_NAME"));
                    column.setPrimaryKey(true);
                    int order = resultSet.getInt("KEY_SEQ");
                    primaryKey.addColumn(column, order);
                    primaryKey.setName(resultSet.getString("PK_NAME"));
                }
                if (logger.isInfoEnabled()) {
                    logger.info("read primaryKey #" + primaryKey);
                }
                if (primaryKey.getName() != null) {
                    table.setPrimaryKey(primaryKey);
                }
                resultSet.close();
            }
        } catch (SQLException ex) {
            if (logger.isErrorEnabled()) {
                logger.error("Error when reading primary keys", ex);
            }
        } finally {
            cleanConnection(connection, null, resultSet);
        }
    }

    @Override
    public void readForeignKeys(DatabaseConfiguration configuration, Schema schema) {
        Connection connection = null;
        ResultSet resultSet = null;

        try {
            connection = getConnection(configuration);
            DatabaseMetaData metaData = connection.getMetaData();
            for (Table table : schema.getTables()) {
                // foreign key
                resultSet = metaData.getExportedKeys(
                        null,
                        getSchemaNameForMetaData(configuration.getSchemaName()),
                        getTableNameForMetaData(table.getName()));
                ForeignKey reference = null;
                while (resultSet.next()) {
                    String importedTableName = resultSet.getString("FKTABLE_NAME");
                    String importedColumnName = resultSet.getString("FKCOLUMN_NAME");
                    String exportedTableName = resultSet.getString("PKTABLE_NAME");
                    String exportedColumnName = resultSet.getString("PKCOLUMN_NAME");
                    if (logger.isInfoEnabled()) {
                        logger.info("reading foreign key #" + importedTableName + "." + importedColumnName
                                + " -> " + exportedTableName + "." + exportedColumnName);
                    }
                    // Primary table
                    Table exportedTable = schema.getTable(exportedTableName);
                    // Primary column
                    Column exportedColumn = exportedTable.getColumn(exportedColumnName);
                    // Foreign table
                    Table importedTable = schema.getTable(importedTableName);
                    // Foreign column
                    Column importedColumn = importedTable.getColumn(importedColumnName);
                    // foreign key name
                    String referenceName = resultSet.getString("FK_NAME");


                    reference = exportedTable.getExportedKey(referenceName);
                    if (reference == null) {
                        reference = new ForeignKey();
                        reference.setName(referenceName);
                        reference.setExportedTable(exportedTable);
                        reference.setImportedTable(importedTable);
                    }
                    else {
                        reference.setName(referenceName);
                    }
                    reference.addReferencedColumn(importedColumn, exportedColumn);

                    importedTable.addImportedKey(reference);
                    exportedTable.addExportedKey(reference);
                }
            }
        } catch (SQLException ex) {
            if (logger.isErrorEnabled()) {
                logger.error("Error when reading foreign keys", ex);
            }
        } finally {
            cleanConnection(connection, null, resultSet);
        }
    }

    @Override
    public void readIndexes(DatabaseConfiguration configuration, Schema schema) {
        Connection connection = null;
        ResultSet resultSet = null;

        try {
            connection = getConnection(configuration);
            DatabaseMetaData metaData = connection.getMetaData();
            for (Table table : schema.getTables()) {
                resultSet = metaData.getIndexInfo(
                        null,
                        getSchemaNameForMetaData(configuration.getSchemaName()),
                        getTableNameForMetaData(table.getName()),
                        false,
                        false);
                while (resultSet.next()) {

                    boolean unique = !resultSet.getBoolean("NON_UNIQUE");

                    String indexName = resultSet.getString("INDEX_NAME");
                    Index index = table.getIndex(indexName);
                    if (index == null) {
                        index = new Index();
                        index.setName(indexName);
                        table.addIndex(index);
                    }
                    index.setUnique(unique);
                    int ordinalPosition = resultSet.getInt("ORDINAL_POSITION");
                    Column column = table.getColumn(resultSet.getString("COLUMN_NAME"));
                    index.addColumn(column, ordinalPosition);

                    String columnSortSequence = resultSet.getString("ASC_OR_DESC");
                    if (columnSortSequence == null) {
                        index.setColumnSortSequence(ColumnSortSequence.NO_SORT);
                    } else if ("A".equals(columnSortSequence)) {
                        index.setColumnSortSequence(ColumnSortSequence.ASC);
                    } else if ("D".equals(columnSortSequence)) {
                        index.setColumnSortSequence(ColumnSortSequence.DESC);
                    }
                }
                resultSet.close();
            }
        } catch (SQLException ex) {
            if (logger.isErrorEnabled()) {
                logger.error("Error when reading indexes", ex);
            }
        } finally {
            cleanConnection(connection, null, resultSet);
        }
    }

    @Override
    public void readColumnsDefaultValue(DatabaseConfiguration configuration, Schema schema) {
    }

    protected static void cleanConnection(Connection connection, Statement statement, ResultSet resultSet) {
        if (resultSet != null) {
            try {
                resultSet.close();
            } catch (SQLException ex) {
                if (logger.isErrorEnabled()) {
                    logger.error("Error when trying close resultSet", ex);
                }
            }
        }
        if (statement != null) {
            try {
                statement.close();
            } catch (SQLException ex) {
                if (logger.isErrorEnabled()) {
                    logger.error("Error when trying close statement", ex);
                }
            }
        }
        if (connection != null) {
            try {
                connection.close();
            } catch (SQLException ex) {
                if (logger.isErrorEnabled()) {
                    logger.error("Error when trying close connection", ex);
                }
            }
        }
    }

    @Override
    public void readUniqueConstraint(DatabaseConfiguration configuration, Schema schema) {

    }

    @Override
    public void readData(DatabaseConfiguration configuration, Schema schema) {
        String filter = configuration.getFilter();
        StringTokenizer tokenizer = new StringTokenizer(filter, ",");
        while (tokenizer.hasMoreTokens()) {
            String token = tokenizer.nextToken();
            Table table = schema.getTable(token);
            if (table != null) {
                final String request = "select * from " + table.getName();
                Connection connection = null;
                Statement statement = null;
                ResultSet resultSet = null;
                try {
                    if (configuration.getUser() != null) {
                        connection = DriverManager.getConnection(configuration.getUrl(), configuration.getUser(), configuration.getPassword());
                    } else {
                        connection = DriverManager.getConnection(configuration.getUrl());
                    }

                    statement = connection.createStatement();
                    resultSet = statement.executeQuery(request);
                    while (resultSet.next()) {
                        DataLine dataLine = new DataLine();
                        for (Column column : table.getColumns()) {
                            Object value = resultSet.getObject(column.getName());
                            dataLine.add(column, value);
                        }
                        table.addLine(dataLine);
                    }

                } catch (SQLException ex) {
                    Logger.getLogger(GenericSchemaReader.class.getName()).log(Level.SEVERE, configuration.getUrl(), ex);
                } finally {
                    cleanConnection(connection, statement, resultSet);
                }
            } else {
                if (logger.isWarnEnabled()) {
                    logger.warn("can't read data for table " + token + ", table isn't in schema");
                }
            }
        }
    }

    protected Connection getConnection(DatabaseConfiguration configuration) throws SQLException {
        if (configuration.getUser() != null) {
            return DriverManager.getConnection(configuration.getUrl(), configuration.getUser(), configuration.getPassword());
        } else {
            return DriverManager.getConnection(configuration.getUrl());
        }
    }
}
